/*
package com.atguigu.juc;


import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.List;

*/
/**
 * @Author MJS
 * @Date 2021/9/4 16:18
 * @Description
 *//*

public class RocketMq {
    public static void main(String[] args) {

    }



    private static void producer() throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        //实例化生产者，实例化时需要指定生产组名
        DefaultMQProducer producer = new DefaultMQProducer("quickstart_producer");
        //设置namesrc地址，有多个的话用";"隔开
        producer.setNamesrvAddr("192.168.30.87");
        //设置重试次数
        producer.setRetryTimesWhenSendFailed(3);
        //启动生产者
        producer.start();
        for(int i=1;i<=100;i++){
            //创建一条消息，指定了消息的主题topic、标签tag、消息的内容
            Message msg = new Message("TopicQuickStart", "TagA", ("Hello RocketMQ "+i).getBytes());
            //发送消息，并设置消息发送超时时间
            SendResult sendResult = producer.send(msg,1000);
            System.out.println(sendResult);
        }
        //关闭生产者，main方法主线程结束，程序终止
        producer.shutdown();
    }

    private static void consumer() throws MQClientException {
    //实例化消费者，实例化时需要指定消费组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("quickstart_consumer");
        //设置namesrc地址
        consumer.setNamesrvAddr("192.168.30.87");
        */
/*
        * 设置Consumer 第一次启动是从队列头部开始消费还是队列尾部开始消费
        * 如果非第一次启动，那么按照上次消费的位置继续消费
        * *//*

        //代表从头部启动
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //设置每次消费消息的最大数量，默认是1，即一条条拉取
        consumer.setConsumeMessageBatchMaxSize(10);
        //设置订阅的消息主题topic和标签tags，这里订阅TopicQuickStart主题下的所有消息，所以会收到上面生产者发送的该主题下标签为TagA的消息
        consumer.subscribe("TopicQuickStart","*");

        //设置为广播消费模式,这样即使是同一个消费组的消费者，也都会收到订阅的所有消息，不会进行均衡消费
        //consumer.setMessageModel(MessageModel.BROADCASTING);

        //注册消费监听
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                //这里的msgs的数量与每次消费消息的最大数量一样
                for (MessageExt msg : msgs) {
                    System.out.println(Thread.currentThread().getName()+"收到消息：topic:"+msg.getTopic()+",tags:"+msg.getTags()+",msg:"+new String(msg.getBody()));
                }
                //回复RocketMQ，这条消息消费成功，如果返回的是ConsumeConcurrentlyStatus.RECONSUME_LATER，即表明消息消费失败，那RocketMQ会对这条消息进行重发操作
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //启动消费者，main方法主线程结束后，程序不会停止，进入阻塞状态，来一条消息就触发一次监听事件
        consumer.start();
        System.out.println("consumer started");
    }
}
*/
